package com.gin.monitor.agg.connector.mysql;

import com.gin.monitor.agg.utils.GenericClassUtil;
import com.gin.monitor.agg.vo.MonitorInfo;
import com.gin.monitor.agg.vo.ViolationInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;

/**
 * 从数据库中读取数据,返回的类型可以是各种各样
 *
 * @author gin
 * @date 2021/4/30
 */
public abstract class AbstractMysqlSource<T> extends RichSourceFunction<T> {

    boolean flag = true;
    Connection conn = null;
    PreparedStatement pst = null;
    ResultSet set = null;
    /**
     * 注意, 需要子类实现父类才能获取到泛型 T
     */
    Class<T> tClass = GenericClassUtil.getSuperClassGenericType(this.getClass());

    @Override
    public void open(Configuration parameters) throws Exception {
        //Class<T> tClass2 = (Class<T>) ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        conn = DriverManager.getConnection("jdbc:mysql://node01/test", "root", "123456");
        if (tClass.getName().equals(MonitorInfo.class.getName())) {
            pst = conn.prepareStatement("select monitor_id,road_id,speed_limit,area_id from t_monitor_info where speed_limit > 0");
        }
        if (tClass.getName().equals(ViolationInfo.class.getName())) {
            pst = conn.prepareStatement(" select car ,violation, create_time from t_violation_list");
        }
    }

    @Override
    public void run(SourceContext<T> ctx) {
        //Class<T> tClass2 = (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        try {
            while (flag) {
                set = pst.executeQuery();
                while (set.next()) {
                    if (tClass.getName().equals(MonitorInfo.class.getName())) {
                        MonitorInfo info = new MonitorInfo(set.getString(1), set.getString(2), set.getInt(3), set.getString(4));
                        ctx.collect((T) info);
                    }
                    if (tClass.getName().equals(ViolationInfo.class.getName())) {
                        ViolationInfo info = new ViolationInfo(set.getString(1), set.getString(2), set.getLong(3));
                        ctx.collect((T) info);
                    }
                }
                Thread.sleep(2000);
                set.close();
            }
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void cancel() {
        flag = false;
    }

    @Override
    public void close() {
        try {
            pst.close();
            conn.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


}
